package com.corgi.reactor;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.TimeUnit;

/**
 * @title: ReactorApplication
 * @description:
 * @author: dengmiao
 * @create: 2019-10-11 11:41
 **/
@Slf4j
@SpringBootApplication
public class ReactorApplication implements ApplicationRunner {

	public static void main(String[] args) {
		SpringApplication.run(ReactorApplication.class, args);
	}

	@Override
	public void run(ApplicationArguments args) throws Exception {
		Flux.range(1, 6)
				.doOnRequest(n -> log.info("Request {} number", n))
				.doOnComplete(() -> log.info("Publisher COMPLETE 1"))
				.publishOn(Schedulers.elastic())
				.map(i -> {
					log.info("Publisher {}, {}", Thread.currentThread(), i);
					// return i;
					return 10 / (i - 3);
				})
				.doOnComplete(() -> log.info("Publisher COMPLETE 2"))
				.subscribeOn(Schedulers.single())
				.onErrorResume(e -> {
					log.error("Exception {}", e.toString());
					return Mono.just(-1);
				})
				/*.onErrorReturn(-1)*/
				.subscribe(i -> log.info("Subscriber {}, {}", Thread.currentThread(), i),
						e -> log.error("error {}", e.toString()),
						() -> log.info("Subscriber COMPLETE")
						, s -> s.request(4)
				);
		TimeUnit.MILLISECONDS.sleep(2000);
	}
}
